# coding: utf-8

"""
CopyRight(c) Xiaodong.Yang, 2018.5, Unidt
将日志数据写入kafka， kafka接收到日志数据后写入HDFS、COS

注意：

    使用该模块的应用，必须保证能够访问到Kafka所在的服务器，否则，
    会导致写操作阻塞。

"""
import datetime
import time

import json
import threading

# from kafka import KafkaProducer

from cache.cache_tools import Cache


class LogData():
    def __init__(self, event='', level='', user='', app='', msg='', time_stamp='20180105 15:24:36'):
        self.event = event
        self.level = level
        self.user = user
        self.app = app
        self.msg = msg
        self.time_stamp = time_stamp

    def to_str(self):
        sep = '\01'
        return self.event + sep + self.level + sep + \
               self.user + sep + self.app + sep + \
               str(self.msg)  + sep + self.time_stamp

    def __repr__(self):
        return self.to_str()

    def __str__(self):
        return self.to_str()

    def encode(self, code='utf-8'):
        return self.to_str().encode(code)


class KafkaLogger(threading.Thread):
    producer = None
    topic = 'framonitor'
    FraCache = Cache()
    FraCache.start()

    def run(self):
        """遍历缓存中的数据，将其写入Kafka"""
        while True:
            """写日志"""
            log = self.FraCache.lpop(self.topic)
            while log is not None:
                # print(log)
                self._send(self.topic, log)
                log = self.FraCache.lpop(self.topic)
            time.sleep(1)

    def set_server(self, **args):
        """
        设置kafka服务器参数
        """
        # self.producer = KafkaProducer(**args)
        return self

    def _send(self, topic, value=None, key=None, partition=None, timestamp_ms=None):
        """向Kafka发送数据"""
        self.producer.send(topic, value, key, partition, timestamp_ms)

    def build_log(self, event='', level='', user='', app='', msg=''):
        """构造日志数据"""
        now = datetime.datetime.now()
        timestamp = now.strftime('%Y%m%d %H:%M:%S')
        data = LogData(event=event, level=level, user=user, app=app, msg=msg, time_stamp=timestamp)
        self.FraCache.rpush(self.topic, data.encode())

    def debug(self, msg='', event='sys', level='debug', app=''):
        """debug log"""
        self.build_log(event=event, level=level, app=app, msg=msg)

    def warning(self, msg='', event='sys', level='warning', app=''):
        """warning log"""
        self.build_log(event=event, level=level, app=app, msg=msg)

    def info(self, msg='', event='sys', level='info', app=''):
        """info log"""
        self.build_log(event=event, level=level, app=app, msg=msg)

    def error(self, msg='', event='sys', level='error', app=''):
        """error log"""
        self.build_log(event=event, level=level, app=app, msg=msg)

    def log_event(self, event='', app='', user='', msg=''):
        self.build_log(event=event, level='app', user=user, app=app, msg=msg)

    def write_log(self, msg='', event='', product_id='', **args):
        """兼容Fralog"""
        user = ''
        if args.get('user') is not None:
            user = args.get('user')
        self.build_log(event=event, level='app', app=product_id, msg=msg, user=user)


if __name__ == '__main__':

    # data = LogData(event='hello', level='debug')
    # print(data)

    # log = KafkaLogger().set_server(bootstrap_servers='192.168.0.14:9092')
    # log.start()
    # time.sleep(0.2)
    # log.debug('hello')
    # time.sleep(0.1)
    # log.warning('hello')
    # time.sleep(0.1)
    # log.info('hello')
    # time.sleep(0.1)
    # log.error('hello')
    # time.sleep(0.1)


    # log.set_server(bootstrap_servers='192.168.0.14:9092')
    # for i in range(10):
    #     log.debug('Hello Kafka')
    #     time.sleep(1)
    # print('sent ok')



